use std::{sync::{atomic::{AtomicU8, Ordering}, Arc}, time::Duration};

use axum::body::Bytes;
use log::error;
use tokio::{io::{split, AsyncReadExt, AsyncWriteExt, BufReader, WriteHalf}, sync::{oneshot, Mutex, Semaphore}, time::timeout};
use tokio_serial::{SerialPortBuilderExt, SerialStream};
use crate::{error::AppErr, global::APP, serialport::proto::{poll_ack, HEAD0, HEAD1}, utils::new_buf};
use super::{ address::MY, dispatch::Router, notify::{handle_log, handle_notify}, proto::{make_empty, xor_sum, Cmd, Frame, Proto, ToProtoBytes, END, TYPE_CONNECT, TYPE_CONNECT_ACK, TYPE_LOG, TYPE_NOTIFY, TYPE_PING, TYPE_POLL, TYPE_POLL_ACK, TYPE_PONG, TYPE_REQ, TYPE_REQ_ACK, TYPE_RES, TYPE_RES_ACK, TYPE_SIMPLE_REQ, TYPE_SIMPLE_RES}, write_port, BodyEncode};

unsafe impl Send for SerialPortConn {}
unsafe impl Sync for SerialPortConn {}

pub struct SerialPortConn {
    w: Mutex<WriteHalf<SerialStream>>,
    shared: Arc<Shared>,
    close_rx: oneshot::Receiver<()>,
    seq: AtomicU8,
}

struct Shared {
    exit_sem: Semaphore,
    router: Router,
}

impl SerialPortConn {

    pub fn new(name: &str, baud_rate: u32) -> Result<Self, AppErr> {
        let stream = tokio_serial::new(name, baud_rate).open_native_async()?;
        let (r, w) = split(stream);
        let reader = BufReader::with_capacity(256, r);
        let shared = Arc::new(Shared {
            exit_sem: Semaphore::const_new(0),
            router: Router::new(),
        });
        
        let (close_tx, close_rx) = oneshot::channel();
        tokio::spawn(read_task(reader, shared.clone(), close_tx));
        let conn = SerialPortConn {
            w: Mutex::const_new(w),
            shared,
            close_rx,
            seq: AtomicU8::new(0),
        };
        Ok(conn)
    }

    fn create_seq(&self) -> u8 {
        self.seq.fetch_add(1, Ordering::Relaxed)
    }

    async fn exec3(&self, ret_ty: u8, seq: u8, dst: u8, buf: &[u8]) -> Result<Frame, AppErr> {
        let mut count: i32 = 0;
        loop {
            let rx = self.shared.router.subscribe_one(dst, ret_ty, seq);
            self.write(buf).await?;
            match timeout(Duration::from_millis(500), rx).await {
                Ok(Ok(frame)) => {
                    return Ok(frame);
                },
                Ok(Err(_)) => {
                    return Err(AppErr::Static("exec3 recv err"));
                }
                Err(_) => {
                    count += 1;
                }
            };
            if count >= 3 {
                return Err(AppErr::Static("通信故障 串口发送重试3次失败"));
            }
        }
    }

    pub async fn simple_req<T: BodyEncode>(&self, dst: u8, cmd: u8, body: T) -> Result<Frame, AppErr> {
        let seq = self.create_seq();
        let buf = Proto {
            ty: TYPE_SIMPLE_REQ,
            seq,
            src: MY,
            dst,
            body: Cmd {
                cmd,
                body
            }
        }.to_bytes();
        self.exec3(TYPE_SIMPLE_RES, seq, dst, &buf).await
    }

    async fn poll(&self, seq: u8, dst: u8) -> Result<u8, AppErr> {
        let buf = make_empty(TYPE_POLL, seq, MY, dst);
        let frame = self.exec3(TYPE_POLL_ACK, seq, dst, &buf).await?;
        Ok( frame.poll_ack_result() )
    }

    async fn connect(&self, seq: u8, dst: u8) -> Result<(), AppErr> {
        let buf = make_empty(TYPE_CONNECT, seq, MY, dst);
        self.exec3(TYPE_CONNECT_ACK, seq, dst, &buf).await?;
        Ok(())
    }

    async fn do_req<T: BodyEncode>(&self, seq: u8, dst: u8, cmd: u8, body: T) -> Result<(), AppErr> {
        let buf = Proto {
            ty: TYPE_REQ,
            seq,
            src: MY,
            dst,
            body: Cmd {
                cmd,
                body
            }
        }.to_bytes();
        self.exec3(TYPE_REQ_ACK, seq, dst, &buf).await?;
        Ok(())
    }

    pub async fn req<T: BodyEncode>(&self, dst: u8, cmd: u8, body: T) -> Result<Frame, AppErr> {
        let seq = self.create_seq();
        self.connect(seq, dst).await?;
        let mut rx = self.shared.router.subscribe_mpsc(dst, TYPE_RES, seq);
        self.do_req(seq, dst, cmd, body).await?;
        let mut complete_flag = false;

        loop {
            match timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(frame)) => {
                    return Ok(frame);
                },
                Ok(None) => {
                    return Err(AppErr::Static("req recv none"));
                },
                Err(_e) => {}
            };
            if complete_flag {
                return Err(AppErr::Static("执行完成 但是没用收到返回结果"));
            }
            let result = self.poll(seq, dst).await?;
            complete_flag = result == poll_ack::COMPLETE
        }
    }

    pub async fn ping(&self, dst: u8) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let buf = make_empty(TYPE_PING, seq, MY, dst);
        self.exec3(TYPE_PONG, seq, dst, &buf).await?;
        Ok(())
    }

    pub async fn close(self) {
        self.shared.exit_sem.add_permits(1);
        _ = self.close_rx.await;
    }

    pub async fn write(&self, buf: &[u8]) -> Result<(), AppErr> {
        self.w.lock().await.write_all(buf).await?;
        Ok(())
    }
}


async fn read_task<R: AsyncReadExt + Unpin>(mut r: R, shared: Arc<Shared>, close_tx: oneshot::Sender<()>) {
    tokio::select! {
        ret = read_loop(&mut r, &shared.router) => {
            if let Err(e) = ret {
                error!("read err:{}", e);
            }
        }
        _ = shared.exit_sem.acquire() => {
            error!("serial read exit");
        }
    }
    APP.notify_info("串口读取退出");
    drop(r);
    _ = close_tx.send(());
}

async fn read_loop<R: AsyncReadExt + Unpin>(r: &mut R, router: &Router) -> Result<(), AppErr> {

    loop {
        let len = sync(r).await?;
        if len < 9 {
            error!("invalid len:{}", len);
            continue;
        }
        
        let mut buf = new_buf(len);
        buf[0] = HEAD0;
        buf[1] = HEAD1;
        buf[2] = len as u8;
        r.read_exact(&mut buf[3..]).await?;

        let sum = xor_sum(&buf[3..(len - 2)]);
        if sum != buf[len - 2] {
            error!("check sum fail");
            continue;
        }

        if END != buf[len - 1] {
            error!("end invalid");
            continue;
        }
        let buf = Bytes::from(buf);

        let frame = Frame::new(buf);
        let ty = frame.ty();
        if ty == TYPE_NOTIFY {
            handle_notify(frame);
            continue;
        }

        if ty == TYPE_LOG {
            handle_log(frame);
            continue;
        }

        if ty == TYPE_RES {
            let buf = make_empty(TYPE_RES_ACK, frame.seq(), frame.dst(), frame.src());
            _ = write_port(&buf).await
        }
        router.notify(frame);
    }
}

async fn sync<R: AsyncReadExt + Unpin>(r: &mut R) -> Result<usize, AppErr> {
    let mut flag = false;
    loop {
        let value = read_byte(r).await?;
        if flag && (value == HEAD1) {
            break;
        }
        flag = value == HEAD0;
    }
    let n = read_byte(r).await?;
    Ok(n as usize)
}

async fn read_byte<R: AsyncReadExt + Unpin>(r: &mut R) -> Result<u8, AppErr> {
    let mut buf = [0u8];
    let n = r.read(&mut buf).await?;
    if n == 0 {
        return Err(AppErr::Static("conn closed?"));
    }
    return Ok(buf[0]);
}
